package com.chenxu.gmall.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.chenxu.gmall.realtime.app.func.DimAsyncFunction;
import com.chenxu.gmall.realtime.bean.OrderDetail;
import com.chenxu.gmall.realtime.bean.OrderInfo;
import com.chenxu.gmall.realtime.bean.OrderWide;
import com.chenxu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Date;
import java.util.concurrent.TimeUnit;

/**
 * Date: 2021/07/15
 * Desc: 合并订单宽表
 * <p>
 * 业务执行流程
 * 1.模拟生成数据jar
 * 2.在MySQL数据库的表中插入相关的业务数据
 * 3.MySQL的Binlog记录变化的数据
 * 4.Maxwell会将变化的数据采集到，并且发送到Kafka的ods_base_db_m
 * 5.BaseDBApp从ods_base_db_m主题中读取数据，进行分流操作
 * -事实数据  发送到kafka的dwd层
 * dwd_order_info
 * dwd_order_detail
 * <p>
 * -维度数据	发送到Hbase的表中
 * DIM_USER_INFO
 * 6.OrderWideApp从kafka的dwd层和Hbase的维度表中读取数据
 * -双流join
 * 使用的是intervalJoin  对OrderInfo和OrderDetail进行join
 * -维度关联
 * 订单表和订单明细表关联起来后，与用户表、地区表、品牌表、品类表、SPU表、SKU表等关联起来；
 */
public class OrderWideApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1  准备本地测试流环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //1.2 设置并行度
        env.setParallelism(1);

        //1.3 设置Checkpoint
        //env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        //env.getCheckpointConfig().setCheckpointTimeout(60000);
        //env.setStateBackend(new FsStateBackend("hdfs://hadoop202:8020/gmall/checkpoint/uniquevisit"))

        //TODO 2.从Kafka的DWD层读取订单和订单明细数据
        //2.1 声明相关的主题以及消费者组
        String orderInfoSourceTopic = "dwd_order_info";
        String orderDetailSourceTopic = "dwd_order_detail";
        String orderWideSinkTopic = "dwm_order_wide";
        String groupId = "order_wide_group";

        //2.2 读取订单主题数据
        FlinkKafkaConsumer<String> orderInfoSource = MyKafkaUtil.getKafkaSource(orderInfoSourceTopic, groupId);
        DataStreamSource<String> orderInfoJsonStrDS = env.addSource(orderInfoSource);

        //2.3 读取订单明细数据
        FlinkKafkaConsumer<String> orderDetailSource = MyKafkaUtil.getKafkaSource(orderDetailSourceTopic, groupId);
        DataStreamSource<String> orderDetailJsonStrDS = env.addSource(orderDetailSource);

        /*
        业务数据大概样式：
        {"database":"gmall0709","table":"order_info","type":"update","ts":1626008658,"xid":254,"xoffset":4651,
        "data":{"id":28993,"consignee":"蒋莲真","consignee_tel":"13494023727","total_amount":21996.00,
        "order_status":"1001","user_id":438,"payment_way":null,"delivery_address":"第17大街第39号楼1单元977门",
        "order_comment":"描述178284","out_trade_no":"352538349474172","trade_body":"十月稻田 长粒香大米 东北大米 东北香米 5kg等10件商品",
        "create_time":"2021-07-11 21:04:17","operate_time":"2021-07-11 21:04:18","expire_time":"2021-07-11 21:19:17",
        "process_status":null,"tracking_no":null,"parent_order_id":null,"img_url":"http://img.gmall.com/646283.jpg",
        "province_id":29,"activity_reduce_amount":0.00,"coupon_reduce_amount":0.00,"original_total_amount":21989.00,
        "feight_fee":7.00,"feight_fee_reduce":null,"refundable_time":null},
        "old":{"operate_time":null}}
        有些数据不全
        data中的字段是我们实际分流后维度的数据流，这在TableProcessFunction中已经做过过滤了；
         */

        //TODO 3.对读取的数据进行结构的转换      jsonString -->OrderInfo|OrderDetail
        //3.1 转换订单数据结构
        SingleOutputStreamOperator<OrderInfo> orderInfoDS = orderInfoJsonStrDS.map(
            new RichMapFunction<String, OrderInfo>() {
                SimpleDateFormat sdf = null;

                @Override
                public void open(Configuration parameters) throws Exception {
                    sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                }

                @Override
                public OrderInfo map(String jsonStr) throws Exception {
                    //直接转化为对应的订单实体类和订单明细实体类；
                    //两个参数分别为JSON类型的字符串和要转化的class类型；
                    OrderInfo orderInfo = JSON.parseObject(jsonStr, OrderInfo.class);
                    //create_ts是没有的，这里给他赋值；create_time的格式为"2021-07-11 21:04:17"；用parse把字符串转化为日期，再通过getTime()转化为时间戳；
                    orderInfo.setCreate_ts(sdf.parse(orderInfo.getCreate_time()).getTime());
                    return orderInfo;
                }
            }
        );

        //3.2 转换订单明细数据结构
        SingleOutputStreamOperator<OrderDetail> orderDetailDS = orderDetailJsonStrDS.map(
            new RichMapFunction<String, OrderDetail>() {
                SimpleDateFormat sdf = null;

                @Override
                public void open(Configuration parameters) throws Exception {
                    sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                }

                @Override
                public OrderDetail map(String jsonStr) throws Exception {
                    OrderDetail orderDetail = JSON.parseObject(jsonStr, OrderDetail.class);
                    orderDetail.setCreate_ts(sdf.parse(orderDetail.getCreate_time()).getTime());
                    return orderDetail;
                }
            }
        );

        //测试
        //➢ 启动 Maxwell、zk、kafka、hdfs、hbase、redis
        //➢ 运行 Idea 中的 BaseDBApp
        //➢ 运行 Idea 中的 OrderWideApp
        //➢ 在数据库 gmall2021_realtime 的配置表中配置订单和订单明细
        //➢ 执行 rt_dblog 下的 jar，生成模拟数据
        //➢ 查看控制台输出
        //流程顺序：脚本--》mysql--》maxwell监控binlog文件--》ods_base_db_m--》BaseDBApp--》dwd_xxx--和hbase》
        //事实数据：dwd_order_info 和 dwd_order_detail->OrderWideApp 从 kafka 的 dwd层读数据，打印输出

        //两个配置文件如下：
        //order_info insert kafka dwd_order_info id,consignee,consignee_tel,total_count,order_status,user_id,payment_way,delivery_address,order_comment,out_trade_no,trade_body,create_time,operate_time,expire_time,process_status,tracking_no,parent_order_id,img_url,province_id,activity_reduce_amount,coupon_reduce_amount,original_total_amount,feight_fee,feight_fee_reduce,refundable_time id null
        //order_detail insert kafka dwd_order_detail id,order_id,sku_id,sku_name,img_url,order_price,sku_num,create_time,source_type,source_id,split_total_amount,split_activity_amount,split_coupon_amount id null

        //orderInfoDS.print("orderInfo>>>");
        //orderDetailDS.print("orderDetail>>>");

        //TODO 4. 指定事件时间字段
        //4.1 订单指定事件时间字段
        SingleOutputStreamOperator<OrderInfo> orderInfoWithTsDS = orderInfoDS.assignTimestampsAndWatermarks(
            WatermarkStrategy
                .<OrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(new SerializableTimestampAssigner<OrderInfo>() {
                    @Override
                    public long extractTimestamp(OrderInfo orderInfo, long recordTimestamp) {
                        return orderInfo.getCreate_ts();
                    }
                })
        );
        //4.2 订单明细指定事件时间字段
        SingleOutputStreamOperator<OrderDetail> orderDetailWithTsDS = orderDetailDS.assignTimestampsAndWatermarks(
                //设定了一个3秒的延迟；
            WatermarkStrategy.<OrderDetail>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(
                    new SerializableTimestampAssigner<OrderDetail>() {
                        @Override
                        public long extractTimestamp(OrderDetail orderDetail, long recordTimestamp) {
                            return orderDetail.getCreate_ts();
                        }
                    }
                )
        );

        //用id进行分组，方便后续双流的join；
        //TODO 5.按照订单id进行分组  指定关联的key
        //注意订单ID在两个表中，一个是id、一个是order_id；
        KeyedStream<OrderInfo, Long> orderInfoKeyedDS = orderInfoWithTsDS.keyBy(OrderInfo::getId);
        KeyedStream<OrderDetail, Long> orderDetailKeyedDS = orderDetailWithTsDS.keyBy(OrderDetail::getOrder_id);

        //Flink中Join包括windowJoin和intervalJoin两种；
        //这里用窗口化的来做明显很难进行窗口的选择，选用间隔的Join方式更简便；
        //TODO 6.使用intervalJoin对订单和订单明细进行关联
        SingleOutputStreamOperator<OrderWide> orderWideDS = orderInfoKeyedDS
            .intervalJoin(orderDetailKeyedDS)
            .between(Time.milliseconds(-5), Time.milliseconds(5))
            .process(
                new ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>() {
                    @Override
                    public void processElement(OrderInfo orderInfo, OrderDetail orderDetail, Context ctx, Collector<OrderWide> out) throws Exception {
                        out.collect(new OrderWide(orderInfo, orderDetail));
                    }
                }
            );

        //测试方法与之前类似；
        //orderWideDS.print("orderWide>>>>");
        //输出数据大致为：orderWide>>>>:4> OrderWide{detail_id=88261, order_id=29895, sku_id=12, order_price=9197.00,
        // sku_num=2, sku_name='Apple iPhone 12 (A2404) 128GB 黑色 支持移动联通电信5G 双卡双待手机', province_id=25,
        // order_status='1001', user_id=226, total_amount=null, activity_reduce_amount=0.00, coupon_reduce_amount=0.00,
        // original_total_amount=36391.00, feight_fee=7.00, split_feight_fee=null, split_activity_amount=null,
        // split_coupon_amount=null, split_total_amount=18394.00, expire_time='null', create_time='2021-07-16 09:40:53',
        // operate_time='null', create_date='null', create_hour='null', province_name='null', province_area_code='null',
        // province_iso_code='null', province_3166_2_code='null', user_age=null, user_gender='null', spu_id=null,
        // tm_id=null, category3_id=null, spu_name='null', tm_name='null', category3_name='null'}
        //有一些数据还是空的，因为维度表还没关联；

        //TODO 7.关联用户维度
        //后续每一步的关联动作都是运用AsyncDataStream.unorderedWait方法；此时订单和订单明细表已经关联好了；
        /*
        ➢ 无序等待（unorderedWait）
        后来的数据，如果异步查询速度快可以超过先来的数据，这样性能会更好一些，但是会有乱序出现。
        ➢ 有序等待（orderedWait）
        严格保留先来后到的顺序，所以后来的数据即使先完成也要等前面的数据。所以性能会差一些
         */
        SingleOutputStreamOperator<OrderWide> orderWideWithUserDS = AsyncDataStream.unorderedWait(
            orderWideDS,
            //构造器中确定具体是和哪个表进行维度关联；然后根据配置表来找到对应的mysql中的表的内容（这里就是user_info)，然后根据id进行关联；
            new DimAsyncFunction<OrderWide>("DIM_USER_INFO") {
                //用id来进行关联，
                @Override
                public String getKey(OrderWide orderWide) {
                    return orderWide.getUser_id().toString();
                }

                @Override
                //JSONObject dimInfoJsonObj是维度表根据id查询到的数据；
                public void join(OrderWide orderWide, JSONObject dimInfoJsonObj) throws Exception {
                    //获取用户生日，数据类似2003-07-09；（user_info表获得）
                    String birthday = dimInfoJsonObj.getString("BIRTHDAY");
                    //定义日期转换工具类
                    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
                    //将生日字符串转换为日期对象
                    Date birthdayDate = sdf.parse(birthday);
                    //获取生日日期的毫秒数
                    Long birthdayTs = birthdayDate.getTime();

                    //获取当前时间的毫秒数
                    Long curTs = System.currentTimeMillis();

                    //年龄毫秒数
                    Long ageTs = curTs - birthdayTs;
                    //转换为年龄
                    Long ageLong = ageTs / 1000L / 60L / 60L / 24L / 365L;
                    Integer age = ageLong.intValue();

                    //将维度中的年龄赋值给订单宽表中的属性
                    orderWide.setUser_age(age);

                    //将维度中的性别赋值给订单宽表中的属性
                    orderWide.setUser_gender(dimInfoJsonObj.getString("GENDER"));
                }
            },
            60, TimeUnit.SECONDS);

        //测试，用户维度；
        //orderWideWithUserDS.print("order----->");
        //查询结果为：
        /*
        order----->:2> OrderWide{detail_id=88664, order_id=30160, sku_id=13, order_price=4188.00, sku_num=1,
        sku_name='华为 HUAWEI P40 麒麟990 5G SoC芯片 5000万超感知徕卡三摄 30倍数字变焦 6GB+128GB亮黑色全网通5G手机', province_id=34,
        order_status='1001', user_id=1959, total_amount=4207.00, activity_reduce_amount=0.00, coupon_reduce_amount=0.00,
        original_total_amount=4188.00, feight_fee=19.00, split_feight_fee=null, split_activity_amount=null,
        split_coupon_amount=null, split_total_amount=4188.00, expire_time='null', create_time='2021-07-16 15:49:31',
        operate_time='null', create_date='null', create_hour='null', province_name='null', province_area_code='null',
        province_iso_code='null', province_3166_2_code='null', user_age=51, user_gender='F', spu_id=null, tm_id=null,
        category3_id=null, spu_name='null', tm_name='null', category3_name='null'}
        //检查发现有user_age=51, user_gender='F'这样的数据，说明维度关联成功；
         */

        //TODO 8.关联省市维度
        SingleOutputStreamOperator<OrderWide> orderWideWithProvinceDS = AsyncDataStream.unorderedWait(
            orderWideWithUserDS,
            //对应表为"base_province"；通过province_id关联；
            new DimAsyncFunction<OrderWide>("DIM_BASE_PROVINCE") {
                @Override
                public String getKey(OrderWide orderWide) {
                    return orderWide.getProvince_id().toString();
                }

                @Override
                public void join(OrderWide orderWide, JSONObject dimInfoJsonObj) throws Exception {
                    orderWide.setProvince_name(dimInfoJsonObj.getString("NAME"));
                    orderWide.setProvince_area_code(dimInfoJsonObj.getString("AREA_CODE"));
                    orderWide.setProvince_iso_code(dimInfoJsonObj.getString("ISO_CODE"));
                    orderWide.setProvince_3166_2_code(dimInfoJsonObj.getString("ISO_3166_2"));
                }
            },
            60,
            TimeUnit.SECONDS
        );

        //orderWideWithProvinceDS.print(">>>>>");

        //TODO 9.关联SKU维度
        SingleOutputStreamOperator<OrderWide> orderWideWithSkuDS = AsyncDataStream.unorderedWait(
            orderWideWithProvinceDS,
            //对应表为"sku_info"；通过sku_id关联；
            new DimAsyncFunction<OrderWide>("DIM_SKU_INFO") {
                @Override
                public String getKey(OrderWide orderWide) {
                    return orderWide.getSku_id().toString();
                }

                @Override
                public void join(OrderWide orderWide, JSONObject dimInfoJsonObj) throws Exception {
                    orderWide.setSku_name(dimInfoJsonObj.getString("SKU_NAME"));
                    orderWide.setSpu_id(dimInfoJsonObj.getLong("SPU_ID"));
                    orderWide.setCategory3_id(dimInfoJsonObj.getLong("CATEGORY3_ID"));
                    orderWide.setTm_id(dimInfoJsonObj.getLong("TM_ID"));
                }
            },
            60,
            TimeUnit.SECONDS
        );
        //orderWideWithSkuDS.print(">>>>");

        //TODO 10.关联SPU商品维度
        SingleOutputStreamOperator<OrderWide> orderWideWithSpuDS = AsyncDataStream.unorderedWait(
            orderWideWithSkuDS,
                //对应表为"spu_info"；通过spu_id关联；
            new DimAsyncFunction<OrderWide>("DIM_SPU_INFO") {
                @Override
                public String getKey(OrderWide orderWide) {
                    return orderWide.getSpu_id().toString();
                }

                @Override
                public void join(OrderWide orderWide, JSONObject dimInfoJsonObj) throws Exception {
                    orderWide.setSpu_name(dimInfoJsonObj.getString("SPU_NAME"));
                }
            },
            60,
            TimeUnit.SECONDS
        );

        //orderWideWithSpuDS.print(">>>>>");

        //TODO 11.关联品类维度
        SingleOutputStreamOperator<OrderWide> orderWideWithCategory3DS = AsyncDataStream.unorderedWait(
            orderWideWithSpuDS,
                //对应表为"base_category3"；通过name关联；
            new DimAsyncFunction<OrderWide>("DIM_BASE_CATEGORY3") {
                @Override
                public void join(OrderWide orderWide, JSONObject jsonObject) throws Exception {
                    orderWide.setCategory3_name(jsonObject.getString("NAME"));
                }

                @Override
                public String getKey(OrderWide orderWide) {
                    //同toString方法也可以；
                    return String.valueOf(orderWide.getCategory3_id());
                }
            }, 60, TimeUnit.SECONDS);

        //orderWideWithCategory3DS.print(">>>>>");

        //TODO 12.关联品牌维度
        SingleOutputStreamOperator<OrderWide> orderWideWithTmDS = AsyncDataStream.unorderedWait(
            orderWideWithCategory3DS,
                //对应表为"base_trademark"；通过getTm_id关联；
            new DimAsyncFunction<OrderWide>("DIM_BASE_TRADEMARK") {
                @Override
                public void join(OrderWide orderWide, JSONObject jsonObject) throws Exception {
                    orderWide.setTm_name(jsonObject.getString("TM_NAME"));
                }

                @Override
                public String getKey(OrderWide orderWide) {
                    return String.valueOf(orderWide.getTm_id());
                }
            }, 60, TimeUnit.SECONDS);

        //测试方法相同；
        //>>>>>:1> OrderWide{detail_id=89087, order_id=30441, sku_id=11, order_price=8197.00, sku_num=1,
        // sku_name='Apple iPhone 12 (A2404) 64GB 白色 支持移动联通电信5G 双卡双待手机', province_id=31, order_status='1001',
        // user_id=1903, total_amount=8286.00, activity_reduce_amount=0.00, coupon_reduce_amount=0.00,
        // original_total_amount=8266.00, feight_fee=20.00, split_feight_fee=null, split_activity_amount=null,
        // split_coupon_amount=null, split_total_amount=8197.00, expire_time='null', create_time='2021-07-16 20:06:47',
        // operate_time='null', create_date='null', create_hour='null', province_name='四川', province_area_code='510000',
        // province_iso_code='CN-51', province_3166_2_code='CN-SC', user_age=18, user_gender='F', spu_id=3, tm_id=2,
        // category3_id=61, spu_name='Apple iPhone 12', tm_name='苹果', category3_name='手机'}
        //可以看到，关联数据都关联好了；

        //kafka中接收到的数据为：
        //{"activity_reduce_amount":0.00,"category3_id":86,"category3_name":"平板电视","coupon_reduce_amount":0.00,
        // "create_time":"2021-07-16 10:53:42","detail_id":89306,"feight_fee":7.00,"order_id":30598,
        // "order_price":2899.00,"order_status":"1001","original_total_amount":5798.00,"province_3166_2_code":"CN-AH",
        // "province_area_code":"340000","province_id":9,"province_iso_code":"CN-34","province_name":"安徽","sku_id":20,
        // "sku_name":"小米电视E65X 65英寸 全面屏 4K超高清HDR 蓝牙遥控内置小爱 2+8GB AI人工智能液晶网络平板电视 L65M5-EA",
        // "sku_num":2,"split_total_amount":5798.00,"spu_id":6,"spu_name":"小米电视 内置小爱 智能网络液晶平板教育电视",
        // "tm_id":5,"tm_name":"小米","total_amount":5805.00,"user_age":49,"user_gender":"M","user_id":2113}
        orderWideWithTmDS.print(">>>>>");

        //TODO 13.将关联后的订单宽表数据写回到kafka的DWM层
        orderWideWithTmDS
            .map(
                    //转化为JSON格式的字符串；
                orderWide->JSON.toJSONString(orderWide)
            )
            .addSink(MyKafkaUtil.getKafkaSink(orderWideSinkTopic));

        env.execute();
    }
}